package com.jzl.canal_consumer.etl.et.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.jzl.canal_consumer.bean.TableBatchMergeSqlEntry;
import com.jzl.canal_consumer.etl.et.rocketmq.connector.AbstractMqCanalConnector;
import com.jzl.canal_consumer.etl.et.rocketmq.connector.CanalRocketMqConnector;
import com.jzl.canal_consumer.etl.et.rocketmq.consumer.impl.DefaultCanalRocketMqConsumer;
import com.jzl.canal_consumer.etl.load.mysql.Loader;
import com.jzl.canal_consumer.etl.load.mysql.SqlExecutor;
import com.jzl.canal_consumer.etl.load.mysql.impl.MysqlLoadImpl;
import com.jzl.canal_consumer.etl.et.rocketmq.transform.MqDataTransform;
import com.jzl.canal_consumer.etl.et.rocketmq.transform.impl.MqMessageTransform;
import com.jzl.canal_consumer.util.MessageEntryUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author ZhenWuWang
 */
@Component
public class CanalRocketMqConsumerFactory
{
    private final MessageEntryUtil messageEntryUtil;

    private final SqlExecutor sqlExecutor;

    @Autowired
    public CanalRocketMqConsumerFactory(MessageEntryUtil messageEntryUtil, SqlExecutor sqlExecutor)
    {
        this.messageEntryUtil = messageEntryUtil;
        this.sqlExecutor = sqlExecutor;
    }

    private CanalRocketMqConnector<FlatMessage> createFlatMessageConnector(String nameServices, String groupName,
                                                                           String... topic)
    {
        return new AbstractMqCanalConnector<FlatMessage>(nameServices, groupName, topic)
        {
            @Override
            protected FlatMessage deserializerMsgData(byte[] data)
            {
                return JSON.parseObject(data, FlatMessage.class);
            }
        };
    }

    private CanalRocketMqConnector<Message> createMessageConnector(String nameServices, String groupName, String... topic)
    {
        return new AbstractMqCanalConnector<Message>(nameServices, groupName, topic)
        {
            @Override
            protected Message deserializerMsgData(byte[] data)
            {
                try
                {
                    return CanalMessageDeserializer.deserializer(data);
                } catch (Exception e)
                {
                    e.printStackTrace();
                }
                return null;
//                return CanalMessageDeserializer.deserializer(data);
            }
        };
    }


    public CanalRocketMqConsumer createCanalRocketMqConsumer(String nameServices, String groupName,
                                                             String[] topic, String schemaName, boolean flatMessage)
    {
        if (flatMessage)
        {
            return null;
        } else
        {
            CanalRocketMqConnector<Message> connector = createMessageConnector(nameServices, groupName, topic);
            MqDataTransform<Message, TableBatchMergeSqlEntry> transform = new MqMessageTransform(messageEntryUtil);
            Loader<TableBatchMergeSqlEntry> loader = new MysqlLoadImpl(sqlExecutor, messageEntryUtil);
            return new DefaultCanalRocketMqConsumer<>(connector, transform, loader, schemaName);
        }
    }

    public List<CanalRocketMqConsumer> canalRocketMqConsumerList(List<String> groupSuffixList, String nameServices,
                                                                 String mac, String schemaName, boolean flatMessage)
    {
        if (groupSuffixList == null || groupSuffixList.isEmpty())
        {
            return Collections.emptyList();
        }
        return groupSuffixList
                .stream()
                .map(groupSuffix -> createCanalRocketMqConsumer(nameServices,
                        mac + groupSuffix, new String[]{mac + groupSuffix}, schemaName, flatMessage))
                .collect(Collectors.toList());
    }
}
